package com.wudl.flink.sql;

import com.wudl.flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * @ClassName : Flink_Sql_Function_UDAF
 * @Description : Flink自定义Udaf 函数  - 求一个平均数
 * @Author :wudl
 * @Date: 2021-08-11 22:55
 */

public class Flink_Sql_Function_UDAF {
    public static void main(String[] args) throws Exception {


        //1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //2. 读取端口中的数据并且转化为javaBean
        SingleOutputStreamOperator<WaterSensor> waterSensorDs = env.socketTextStream("192.168.1.180", 9999)
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                });

        // 3. 讲流 转化为动态表
        Table table = tableEnv.fromDataStream(waterSensorDs);
        // 5. 先注册在使用
        tableEnv.createTemporarySystemFunction("mySpilt",mySpilt.class);
        //5.1 在使用注册的自定义函数 名称为MyLength
//        table.joinLateral(call("mySpilt",$("id")))
//                .select($("id"),$("s")).execute().print();
        // 5.1 采用sql 的方式进行使用自定义函数
            tableEnv.sqlQuery("select id, s from  "+table+", lateral table(mySpilt(id))").execute().print();
        //5. 执行任务
        env.execute();


    }

    // 自定义函数类
    @FunctionHint(output = @DataTypeHint("ROW<s STRING>"))
    public static class  mySpilt extends TableFunction<Row> {
        public void eval(String value) {
            String[] split = value.split("_");
            for (String str:split)
            {
                collect(Row.of(str));
            }
        }
    }

}
